rclcpp WaitSet 与 Executor：等待机制与调度分层

大多数 ROS2 节点只需要 rclcpp::spin(node) 一行就能跑起来，但这行代码背后是一条「等待事件 → 执行回调」的循环。rclcpp 把这条循环拆成了两层：底层的 WaitSet 负责回答「现在有没有工作可做」，上层的 Executor 负责决定「工作以什么顺序、在几个线程里执行」。本文拆解这两层的机制、职责边界与配合方式，适合需要自定义执行逻辑、排查回调不触发或并发问题、或者在多线程 Executor 与回调组之间做选型决策的读者。Node 侧的通信对象与回调组 API 可另见本站《rclcpp::Node 类 API 速查》。

WaitSet：底层等待原语

WaitSet 是一个多路复用等待机制，能够同时监听多个 ROS2 实体，直到其中某个实体就绪。

可监听的实体类型：

• Subscription（订阅者）：消息到达时触发
• GuardCondition（守卫条件）：手动触发或内部事件触发
• Timer（定时器）：到期触发
• Client（客户端）：服务响应到达
• Service（服务端）：服务请求到达
• Waitable（可等待对象）：自定义可等待实体

三种配置

rclcpp 通过策略模板组合出三种 WaitSet（定义见 wait_set.hpp）：

// 1. 标准配置：动态存储 + 非线程安全（最常用）
using WaitSet = rclcpp::WaitSetTemplate<
  rclcpp::wait_set_policies::SequentialSynchronization,
  rclcpp::wait_set_policies::DynamicStorage
>;

// 2. 静态配置：固定容量 + 构造后不可变（性能最优）
template<size_t NSub, size_t NGuard, size_t NTimer, ...>
using StaticWaitSet = rclcpp::WaitSetTemplate<
  rclcpp::wait_set_policies::SequentialSynchronization,
  rclcpp::wait_set_policies::StaticStorage<NSub, NGuard, NTimer, ...>
>;

// 3. 线程安全配置：动态存储 + 线程安全
using ThreadSafeWaitSet = rclcpp::WaitSetTemplate<
  rclcpp::wait_set_policies::ThreadSafeSynchronization,
  rclcpp::wait_set_policies::DynamicStorage
>;

标准配置可动态增删实体但非线程安全；StaticWaitSet 构造后不可修改，容量在编译期固定，避免了运行时动态分配；ThreadSafeWaitSet 在标准配置基础上增加了线程安全的增删与 wait 保护，带有一定开销。

工作流程

创建 WaitSet → 添加实体 → wait() 阻塞等待 → 返回就绪实体 → 执行处理 → 重复

关键方法（wait_set_template.hpp）：

• add_*() / remove_*() 系列：添加、移除不同类型的实体
• wait(timeout)：等待任意实体就绪
• prune_destroyed_entities()：清理已销毁的实体

Executor：高层调度器

Executor 是 ROS2 的事件循环调度器，职责包括：管理节点和回调组、使用 WaitSet 等待事件、决策并执行回调、控制执行顺序和并发。

// executor.hpp 核心接口
class Executor {
  virtual void spin() = 0;  // 阻塞式旋转
  void add_callback_group(...);
  void add_node(...);
  void spin_once_impl(...);
};

SingleThreadedExecutor：单线程执行器

class SingleThreadedExecutor : public rclcpp::Executor {
public:
  void spin() override;  // 单线程顺序执行所有回调
};

特点：执行顺序可预测，无并发问题；资源占用少。适合回调执行顺序至关重要的场景、资源受限的嵌入式系统、需要确定性行为的实时系统；缺点是无法利用多核并行。

MultiThreadedExecutor：多线程执行器

explicit MultiThreadedExecutor(
  const rclcpp::ExecutorOptions & options = rclcpp::ExecutorOptions(),
  size_t number_of_threads = 0,  // 0 表示使用 CPU 核心数（最少 2）
  bool yield_before_execute = false,
  std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1)
);

void spin() override;  // 多线程并发执行

number_of_threads 默认为 0，此时按检测到的 CPU 核心数创建线程池，且不少于 2 个（头文件注释原文："the default 0 will use the number of cpu cores found (minimum of 2)"）。多线程执行器能充分利用多核、支持 Reentrant 回调组的并发执行，适合 CPU 密集型回调与高吞吐量应用；代价是需要处理线程同步，执行顺序不可预测。

Executor 内部工作原理

核心成员

class Executor {
  std::atomic_bool spinning;                              // 旋转状态标志
  std::shared_ptr<GuardCondition> interrupt_guard_condition_;  // 中断守卫条件
  std::shared_ptr<GuardCondition> shutdown_guard_condition_;   // 关闭守卫条件
  std::mutex mutex_;                                      // 保护共享状态
  std::shared_ptr<Context> context_;                      // ROS2 上下文

  rclcpp::WaitSet wait_set_;                              // 内部 WaitSet
  std::shared_ptr<executors::ExecutorNotifyWaitable> notify_waitable_;
  std::atomic_bool entities_need_rebuild_;                // 是否需要重建实体集
};

spin 主循环（示意）

void Executor::spin() {
  spinning = true;
  try {
    while (spinning) {
      collect_entities();                        // 收集所有需要监听的实体
      wait_for_work(timeout);                    // 用 WaitSet 等待实体就绪
      while (get_next_ready_executable(any_executable)) {
        execute_any_executable(any_executable);  // 获取并执行就绪的回调
      }
    }
  } catch (...) {
    spinning = false;
    throw;
  }
}

实体重建机制

Executor 通过 notify_waitable_ 监听实体变化。调用 add_node() 或 add_callback_group() 时标记 entities_need_rebuild_ 并触发 interrupt_guard_condition_，唤醒正在 wait_for_work() 中阻塞的循环，重新收集实体并重建 WaitSet。这解释了运行期新增节点后回调能被及时调度。

两层如何配合

┌─────────────────────────────────────────────┐
│ 应用层：rclcpp::spin(node) 或 executor.spin() │
├─────────────────────────────────────────────┤
│ Executor：管理节点与回调组，决策执行顺序与并发 │
├─────────────────────────────────────────────┤
│ WaitSet：多路复用监听实体就绪，封装 rcl_wait   │
├─────────────────────────────────────────────┤
│ rcl / rmw 层：rcl_wait_set_t、DDS 通信       │
└─────────────────────────────────────────────┘

维度	WaitSet	Executor
抽象层次	底层等待原语	高层调度器
关注点	何时有工作	如何执行工作
功能	多路复用等待、实体管理	回调执行、顺序控制、并发管理
使用场景	自定义执行逻辑	绝大多数 ROS2 应用
线程安全	可选（ThreadSafeWaitSet）	MultiThreadedExecutor

使用示例

标准用法：通过 Executor

#include <rclcpp/rclcpp.hpp>

int main(int argc, char ** argv) {
  rclcpp::init(argc, argv);
  auto node = rclcpp::Node::make_shared("my_node");

  auto subscription = node->create_subscription<std_msgs::msg::String>(
    "topic", 10,
    [](const std_msgs::msg::String::SharedPtr msg) {
      RCLCPP_INFO(rclcpp::get_logger("callback"), "Received: '%s'", msg->data.c_str());
    });

  // 方式 1：全局默认 executor（单线程）
  rclcpp::spin(node);

  // 方式 2：显式 SingleThreadedExecutor
  rclcpp::executors::SingleThreadedExecutor executor;
  executor.add_node(node);
  executor.spin();

  // 方式 3：MultiThreadedExecutor，2 个工作线程
  rclcpp::executors::MultiThreadedExecutor executor2(2);
  executor2.add_node(node);
  executor2.spin();

  rclcpp::shutdown();
  return 0;
}

高级用法：直接使用 WaitSet

rclcpp::WaitSet wait_set;
wait_set.add_subscription(subscription);
wait_set.add_timer(timer);

while (rclcpp::ok()) {
  auto result = wait_set.wait(std::chrono::seconds(1));

  if (result.kind() == rclcpp::WaitResultKind::Ready) {
    if (result.is_ready(subscription)) {
      // 自行取消息并执行处理
    }
    if (result.is_ready(timer)) {
      timer->execute_callback();
    }
  } else if (result.kind() == rclcpp::WaitResultKind::Timeout) {
    // 超时，无就绪实体
  } else if (result.kind() == rclcpp::WaitResultKind::Empty) {
    // WaitSet 为空
  }
}

直接操作 WaitSet 时消息的获取与回调执行都要自行处理，只在需要完全自定义执行逻辑时才有必要。

选型与实践

场景	推荐	理由
简单节点	rclcpp::spin(node)	默认单线程，简单可靠
需要控制执行顺序	SingleThreadedExecutor	顺序执行，可预测
CPU 密集型回调	MultiThreadedExecutor	并发处理，提高吞吐量
多个独立任务	MultiThreadedExecutor + Reentrant 回调组	充分利用多核
嵌入式/资源受限	SingleThreadedExecutor	资源占用少
完全自定义执行逻辑	直接使用 WaitSet	灵活控制等待与执行

注意事项：

1. 避免回调阻塞：无论单线程还是多线程 Executor，阻塞回调都会拖慢整体调度
2. Reentrant 回调组：MultiThreadedExecutor 中只有 Reentrant 类型的回调组才能并发执行；默认的 MutuallyExclusive 回调组内仍是串行
3. 回调组归属：一个回调组只能关联一个 Executor，重复关联会抛异常
4. 资源清理：正确调用 rclcpp::shutdown() 释放资源
5. 异常处理：在回调中捕获异常，避免传播到 Executor 主循环

常见问题

为什么 MultiThreadedExecutor 没有带来加速？ 通常有三个原因：回调组类型是默认的 MutuallyExclusive，无法并发执行；回调之间存在互斥锁等同步；任务受 I/O 限制而非 CPU 限制。解法是把需要并发的回调放进 Reentrant 回调组：

auto callback_group = node->create_callback_group(
  rclcpp::CallbackGroupType::Reentrant
);

如何优雅地停止 Executor？ 在另一个线程调用 executor.cancel() 设置 spinning = false；或调用 rclcpp::shutdown() 关闭整个上下文，触发 shutdown 守卫条件唤醒等待。

Executor 如何感知新添加的节点？ 通过 notify_waitable_ 监听实体变化，add_node() 触发 interrupt_guard_condition_ 唤醒等待中的 wait_for_work()，随后重建 WaitSet。

要点

• WaitSet 处理「何时有工作」，Executor 处理「如何执行工作」；两层分工，绝大多数应用只接触 Executor
• Executor 内部持有并维护一个 WaitSet，运行期节点增删通过守卫条件触发重建
• StaticWaitSet 固定容量避免动态分配，适合性能敏感且实体数量已知的场景；ThreadSafeWaitSet 以开销换多线程增删安全
• MultiThreadedExecutor 的并发收益取决于回调组类型：默认 MutuallyExclusive 下仍是串行，需要并发的回调要显式放入 Reentrant 回调组

参考资料

• wait_set.hpp / wait_set_template.hpp（rclcpp 源码）
• executor.hpp（rclcpp 源码）
• ROS 2 Executor Design（官方设计文档）
• DeepWiki: rclcpp 执行器与并发章节